package org.zachary.imitation.raft;

import com.google.protobuf.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zachary.imitation.raft.core.ReplicatorGroup;
import org.zachary.imitation.raft.core.ReplicatorGroupImpl;
import org.zachary.imitation.raft.core.ReplicatorType;
import org.zachary.imitation.raft.entity.Ballot;
import org.zachary.imitation.raft.entity.LogId;
import org.zachary.imitation.raft.error.RaftError;
import org.zachary.imitation.raft.option.NodeOptions;
import org.zachary.imitation.raft.option.RaftMetaStorageOptions;
import org.zachary.imitation.raft.option.RaftOptions;
import org.zachary.imitation.raft.option.ReplicatorGroupOptions;
import org.zachary.imitation.raft.rpc.RpcRequestClosure;
import org.zachary.imitation.raft.rpc.RpcRequests;
import org.zachary.imitation.raft.rpc.RpcResponseClosureAdapter;
import org.zachary.imitation.raft.storage.RaftMetaStorage;
import org.zachary.imitation.raft.util.timer.HashedWheelTimer;
import org.zachary.imitation.raft.util.timer.Repeated;
import org.zachary.utils.concurrent.NamedThreadFactory;
import org.zachary.utils.lang.Asserts;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 整个框架中最核心的类，关于节点的一切功能几乎都定义在该类中了，不是在该类中直接实现就是由该类的某个方法做了一层代理，总之，都会通过这个类的对象去执行操作。
 *
 * Created by guozhaokun on 2024/8/30 16:51.
 */
public class NodeImpl implements Node {
	private Logger logger = LoggerFactory.getLogger(NodeImpl.class);

	/** 当前节点的状态 */
	private volatile State state;
	/** 节点所属 group（集群）以及节点自身信息 */
	private NodeId nodeId;
	/** 记录当前节点所在集群的 Leader 的节点信息，该 Leader 所属的 group 一定和当前节点所属的 group 是一样的 */
	private PeerId leader;
	/** 节点的当前任期 */
	private long term;
	/** 记录当前节点在当前任期中为哪个节点投了票（每个节点在一个任期中只能投一票），被投票的节点一定和当前节点有相同的 group */
	private PeerId votedId;
	/** 当前节点需要了解整个集群的信息，以便和其它节点进行通信 */
	private Configuration conf;
	/** 节点的配置参数 */
	private NodeOptions nodeOptions;
	/** raft 框架的全局配置参数 */
	private RaftOptions raftOptions;
	/** 用于创建各个组件的服务工厂 */
	private JRaftServiceFactory serviceFactory;
	/** 节点元数据存储器 */
	private RaftMetaStorage metaStorage;
	/** 上一次收到 Leader 心跳信息的时间 */
	private volatile long lastLeaderTimestamp = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());;

	/** 超时选举定时器：在规定的时间内未收到来自 Leader 的心跳，则会进行 Leader 选举的预投票 */
	private Repeated electionRepeated;
	/** 计算预投票结果的计票器 */
	private final Ballot prevVoteCtx = new Ballot();

	/** 选举超时定时器：预投票成功后开始正式 Leader 选举，在规定的时间内仍未选举出 Leader，则会进入下一轮 Leader 选举 */
	private Repeated voteRepeated;
	/** 计算正式投票结果的计票器 */
	private final Ballot voteCtx = new Ballot();

	private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
	protected final Lock writeLock = this.readWriteLock.writeLock();
	protected final Lock readLock = this.readWriteLock.readLock();

	/** 复制器组，维护了各个 Follower 节点对应的 Replicator 对象，用于 Leader 向各个 Follower 发送心跳和复制日志 */
	private ReplicatorGroup replicatorGroup;
	/** 全局的定时任务管理器 */
	private ScheduledExecutorService scheduledExecutorService;

	// region getter/setter
	@Override
	public String getGroupId() {
		return nodeId.getGroupId();
	}

	public State getState() {
		return state;
	}

	public void setState(State state) {
		this.state = state;
	}

	@Override
	public NodeId getNodeId() {
		return nodeId;
	}

	public void setNodeId(NodeId nodeId) {
		this.nodeId = nodeId;
	}

	@Override
	public PeerId getLeader() {
		return leader;
	}

	public void setLeader(PeerId leader) {
		this.leader = leader;
	}

	@Override
	public long getTerm() {
		return term;
	}

	public void setTerm(int term) {
		this.term = term;
	}

	public Configuration getConf() {
		return conf;
	}

	public void setConf(Configuration conf) {
		this.conf = conf;
	}
	// endregion
	public NodeImpl(String group, PeerId serverId) {
		this.nodeId = new NodeId(group, serverId);
		// 节点刚刚创建时的状态为未初始化
		this.state = State.UNINITIALIZED;
	}

	// region 初始化节点
	@Override
	public boolean init(NodeOptions options) {
		Asserts.notNull(options, "Null node options");
//		Asserts.notNull(options.getRaftOptions(), "Null raft options");
		Asserts.notNull(options.getServiceFactory(), "Null jraft service factory");

		this.nodeOptions = options;
		this.serviceFactory = options.getServiceFactory();

		// 设置当前节点选举优先级
		this.nodeId.getPeerId().setPriority(options.getElectionPriority());
		if (PeerId.IP_ANY.equals(this.getNodeId().getPeerId().getEndpoint().getIp())) {
			logger.error("Node can't started from IP_ANY.");
			return false;
		}

		// 校验当前节点的 ip:port 是否已经添加到节点管理器中了。节点的网络地址已经在 RaftGroupService#start 中被添加到节点管理器中了
		if (!NodeManager.getInstance().serverExists(this.nodeId.getPeerId().getEndpoint())) {
			logger.error("No RPC server attached to, did you forget to call addService?");
			return false;
		}

		// 创建一个全局的定时任务管理器，在第一版本，这个定时任务管理器只负责启动 Leader 向 Follower 发送心跳消息的任务
		this.scheduledExecutorService = new ScheduledThreadPoolExecutor(this.nodeOptions.getTimerPoolSize(), new NamedThreadFactory("JRaft-Node-ScheduleThreadPool"));

		// 初始化超时选举定时器
		String name = "JRaft-ElectionTimer";
		// 创建超时选举定时器，RepeatedTimer 构造函数的第三个参数创建的是一个时间轮对象，时间轮的刻度是 1 秒
		this.electionRepeated = new Repeated(name, (int) this.nodeOptions.getElectionTimeout().toMillis(), new HashedWheelTimer(new NamedThreadFactory(name, true), 1, TimeUnit.MILLISECONDS, 2048)) {
			@Override
			protected void onTrigger() {
				// 这是超时后需要执行的业务逻辑，就是在该方法中，开始执行超时选举任务
				handleElectionTimeout();
			}

			@Override
			protected int adjustTimeout(final int timeoutMs) {
				// 在一定范围内返回一个随机的时间，这就意味着每个节点的超时选举时间是不同的，否则多个节点同时成为候选者，很可能选举失败
				// return ThreadLocalRandom.current().nextInt(timeoutMs, timeoutMs + this.raftOptions.getMaxElectionDelayMs());
				return ThreadLocalRandom.current().nextInt(timeoutMs, timeoutMs + 600);
			}
		};

		// 初始化选举超时定时器
		name = "JRaft-VoteTimer";
		// 创建选举超时定时器，RepeatedTimer 构造函数的第三个参数创建的是一个时间轮对象，时间轮的刻度是 1 秒
		this.voteRepeated = new Repeated(name, (int) this.nodeOptions.getElectionTimeout().toMillis(), new HashedWheelTimer(new NamedThreadFactory(name, true), 1, TimeUnit.MILLISECONDS, 2048)) {
			@Override
			protected void onTrigger() {
				// 这是超时后需要执行的业务逻辑，就是在该方法中，开始执行选举超时任务
				handleVoteTimeout();
			}

			@Override
			protected int adjustTimeout(final int timeoutMs) {
				// 在一定范围内返回一个随机的时间，这就意味着每个节点的超时选举时间是不同的，否则多个节点同时成为候选者，很可能选举失败
				// return ThreadLocalRandom.current().nextInt(timeoutMs, timeoutMs + this.raftOptions.getMaxElectionDelayMs());
				return ThreadLocalRandom.current().nextInt(timeoutMs, timeoutMs + 600);
			}
		};

		// 初始化节点元数据存储器组件，在初始化的过程中，会把本地元数据文件加载到内存（NodeImpl 的相关字段中）
		if (!initMetaStorage()) {
			// 初始化失败则记录错误信息
			logger.error("Node {} initMetaStorage failed.", getNodeId());
			return false;
		}


		//先创建一个复制器组对象
		this.replicatorGroup = new ReplicatorGroupImpl();
		// TODO 初始化RPC客户端服务
		// this.rpcService = new DefaultRaftClientService(this.replicatorGroup, this.options.getAppendEntriesExecutors());
		//创建一个复制器组需要的配置参数对象
		final ReplicatorGroupOptions replicatorGroupOptions = new ReplicatorGroupOptions();
		// 设置心跳超时时间，即当一个节点成为 Leader 后，经过这个时间之后，就要立刻向 Follower 发送心跳消息了。这个时间一定要比超时选举的时间小，否则就触发选举了
		replicatorGroupOptions.setHeartbeatTimeoutMs((int) Math.max(this.nodeOptions.getElectionTimeout().toMillis() / this.raftOptions.getElectionHeartbeatFactor(), 10));
		//设置超时选举时间
		replicatorGroupOptions.setElectionTimeoutMs(this.nodeOptions.getElectionTimeout().toMillis());
		replicatorGroupOptions.setNode(this);
		// replicatorGroupOptions.setRaftRpcClientService(this.rpcService);
		replicatorGroupOptions.setRaftOptions(this.raftOptions);
		// 设置全局的定时任务管理器
		replicatorGroupOptions.setScheduledExecutorService(this.scheduledExecutorService);
		// TODO 设置性能监控对象
		// this.options.setMetricRegistry(this.metrics.getMetricRegistry());
		// TODO 初始化客户端服务组件，初始化的过程中，创建了一个业务线程池来处理业务逻辑，毕竟 Netty 的单线程执行器只负责处理 IO 逻辑
//		if (!this.rpcService.init(this.options)) {
//			logger.error("Fail to init rpc service.");
//			return false;
//		}
		// 初始化复制器组
		this.replicatorGroup.init(this.nodeId, replicatorGroupOptions);

		// raft 算法中，节点都是很以 Follower 的身份参与进来的，所以在初始化方法的尾部，该方法会修改节点状态为 FOLLOWING、重置 Leader、并启动超时选举定时器
		stepDown(this.term, false, null);

		return true;
	}
	// endregion

	@Override
	public void shutdown() {

	}

	/** 初始化元数据存储器的方法 */
	private boolean initMetaStorage() {
		// 通过服务工厂创建元数据存储器
		this.metaStorage = this.serviceFactory.createRaftMetaStorage(this.nodeOptions.getRaftMetaUri(), null);
		// 创建 RaftMetaStorageOptions 对象，这个对象是专门给元数据存储器使用的，封装了元数据存储器需要的配置参数
		RaftMetaStorageOptions metaStorageOptions = new RaftMetaStorageOptions();
		metaStorageOptions.setNode(this);

		// 初始化元数据存储器，初始化的过程中，会把元数据本地文件中的任期和为谁投票这些数据加载到内存中，实际上就是赋值给 NodeImpl 对应的字段中
		if (!this.metaStorage.init(metaStorageOptions)) {
			logger.error("Node {} init meta storage failed, uri={}.", this.nodeId.getPeerId(), this.nodeOptions.getRaftMetaUri());
			return false;
		}

		// 给当前节点的任期赋值
		this.term = this.metaStorage.getTerm();
		// 得到当前节点的投票记录，这里的 copy 在功能上就相当于一个深拷贝方法
//		this.votedId = this.metaStorage.getVotedFor().copy();

		return true;
	}

	// region 超时选举预投票
	/** 定时器的时间一到就会由 onTrigger 执行到这个方法。当调用这个方法时，不一定会真的进行 Leader 选举。这里要判断当前时间距离上一次收到 Leader 的心跳时间是否超过了规定的间隔，当超过后才会真的开始 Leader 选举 */
	private void handleElectionTimeout() {
		// 是否释放写锁
		boolean doUnlock = true;
		this.writeLock.lock();
		try {
			// 如果当前节点是 Leader，则无需执行超时选举（但定时器还是会启动）
			// 如果当前节点是 Candidate，也无需执行超时选举。因为当正式开始选举时（该方法的尾部），会将自己从 Follower 变更为 Candidate，已经开始选举了
			// 只有当前节点是 Follower（NodeImpl#init 初始化的尾部会将状态由 Uninitialized 变更为 Following，raft 规定节点都是以 Follower 参与进来的），才会定时判断是否长时间未收到心跳，从而开始真正的选举
			if (this.state != State.FOLLOWING) {
				return;
			}

			// 这个判断非常重要，超时选举定时器启动之后，只要节点状态是 Follower，就会一直工作。每过一段时间就会触发一次，但是触发了并不意味着就要真的去进行选举，
			// 会在这里进行判断距离当前节点最后一次收到 Leader 信息过去的多少时间，如果这个时间超过了规定的阈值，才会进入预投票阶段，否则直接退出，这里返回 true 代表还没有超时
			if (isCurrentLeaderValid()) {
				return;
			}

			// 进入选举，说明此时还没有 Leader，需要置当前节点所持有的 Leader 信息为 null
			updateLeaderId(null, null);

			doUnlock = false;
			// 开始预投票
			// 预投票方法除了在 handleElectionTimeout() 中被调用，还会在其它地方调用。进入 preVote 方法前的上层方法一定是已经获取到了写锁的（即预投票的入口还持有着写锁），preVote 方法内部完成后会释放写锁，那么上层方法的 finally 就不需要释放写锁了
			preVote();
		} finally {
			if (doUnlock) {
				this.writeLock.unlock();
			}
		}
	}

	/** 判断 Leader 是否能正常发送心跳给 Follower。true 标识 Leader 正常发送了心跳给 Follower */
	private boolean isCurrentLeaderValid() {
		// 当前时间 - 当前节点最后一次收到 Leader 心跳的时间 < 超时选举时间，则说明这个 Leader 是活跃的，是合法的。否则任务 Leader 不合法了，可以进行超时选举了
		return TimeUnit.NANOSECONDS.toMillis(System.nanoTime()) - this.lastLeaderTimestamp < this.nodeOptions.getElectionTimeout().toMillis();
	}

	/** 修改当前节点记录的 Leader 信息 */
	private void updateLeaderId(final PeerId newLeaderId, Status status) {
		this.leader = newLeaderId;
	}

	/** 在指定的任期中执行身份降级，即，将状态变更为 FOLLOWING，并启动超时选举定时器 */
	private void stepDown(final long term, final boolean wakeupCandidate, final Status status) {
		// 当前节点关闭或出现了异常，直接返回
		if (!this.state.isActive()) {
			return;
		}

		// 正要开始进行 Leader 选举，集群中还没有 Leader，需要把当前节点记录的 Leader 信息置 null
		updateLeaderId(null, null);

		// 当前状态为 Follower 的节点长时间没有收到 Leader 心跳，此时将状态变更为 Candidate，并开始真正的 Leader 选举
		this.state = State.FOLLOWING;

		// 修改当前节点最后一次收到 Leader 心跳的时间为当前时间。只有 Follower 需要启动超时选举定时器，当某个节点率先进行超时选举了，该节点就会变成 Candidate，Candidate 在和其它节点通信时，
		// 其它节点发现自己的任期低于接收到的 Candidate 的任期，就会调用 stepDown 方法将自己的状态变更为 Follower，同时更新接收到心跳的时间戳，这是很合理的。
		// 但是，如果其它 Follower 一直不更新这个时间戳，那么其它 Follower 也会触发超时选举并进行 Leader 选举了。只有更新了这个时间戳，其它 Follower 才会发现还没有触发选举超时，就不会进入超时选举阶段
		this.lastLeaderTimestamp = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());

		// 如果方法参数中的任期比当前节点的任期大，意味着当前节点正在进行投票工作
		if (term > this.term) {
			// 修改当前节点的任期
			this.term = term;
			// 把当前节点的投票记录置空，因为新的任期开始了，一个任期只能投一票。由于新的任期中还没有开始投票，所以先设置为空
			this.votedId = null;
			// 持久化元数据信息
			this.metaStorage.setTermAndVotedFor(term, this.votedId);
		}
		// 启动超时选举定时器
		this.electionRepeated.restart();
	}

	/** 当前节点（候选者）向其它节点发起预投票请求，会携带自增的 term（本地的 term 不变），当前节点的最大日志的索引 */
	private void preVote() {
		// 记录开始预投票的日志
		logger.info("Node {} term {} start preVote.", getNodeId(), this.term);
		try {
			// 当前节点不是集群中的节点，直接退出
			if (!this.conf.contains(this.nodeId.getPeerId())) {
				logger.warn("Node {} can't do preVote as it is not in conf <{}>.", getNodeId(), this.conf);
				return;
			}
		} finally {
			// 如果 当前节点不在集群中，这里一定要释放写锁，因为上层代码已经获取到了写锁，但上层代码的 finally 又不会释放写锁
			this.writeLock.unlock();
		}

		// 是否释放写锁
		boolean doUnlock = true;
		// 重新获取写锁
		this.writeLock.lock();
		try {
			// 定义一个变量记录当前节点任期
			long oldTerm = this.term;

			// 获得当前节点最后一条日志，这里暂时写死，后面会补全完整的代码
			final LogId lastLogId = new LogId(0,0);

			// 初始化预投票计数器，prevVoteCtx 是类型为 Ballot 变量。当前节点收到的投票结果都会被 prevVoteCtx 成员变量收集，如果收到的预投票的票数大于半数，就会进入正式投票状态
			// 初始化预投票计算器就是把集群配置中的节点信息交给 Ballot prevVoteCtx 对象，这样预投票计算器就知道集群中有多少节点了，也知道得到的预投票的票数超过多少就可以进入正式投票状态了
			this.prevVoteCtx.init(this.conf, this.conf);

			// 遍历集群配置中的节点，当前节点依次向这些节点发送预投票请求
			for (final PeerId peer : this.conf.getPeers()) {
				// 如果遍历到的节点就是当前节点，就跳过这次循环
				if (peer.equals(this.nodeId.getPeerId())) {
					continue;
				}

				// 连接遍历到的这个节点，如果不能成功连接，就记录日志
	//			if (!this.rpcService.connect(peer.getEndpoint())) {
	//				logger.warn("Node {} channel init failed, address={}.", getNodeId(), peer.getEndpoint());
	//				continue;
	//			}
				// 创建一个回调对象，这个对象中实现了一个回调函数，该回调函数会在当前节点接收到预投票请求的响应后被回调，可以看出都是异步请求
				// 构造方法中的第一个参数表示要把预投票请求发送到的目标节点，第二个参数是当前节点发送预投票请求时当前节点的本地任期
				final OnPreVoteRpcDone done = new OnPreVoteRpcDone(peer, this.term);

				// 向被遍历到的这个节点发送一个预投票请求，下面就是构建预投票请求的内容
				done.request = RpcRequests.RequestVoteRequest.newBuilder()
						.setPreVote(true) // 设置为 true 意味着是预投票请求
						.setGroupId(this.nodeId.getGroupId()) // 设置当前节点所属集群的id
						.setServerId(this.nodeId.getPeerId().toString()) // 设置当前节点的 ip:port
						.setPeerId(peer.toString()) // 设置发送的目标
						.setTerm(this.term + 1) // 在发送请求时发送的任期值为当前节点的任期值加 1
						.setLastLogIndex(lastLogId.getIndex()) // 当前节点最后一条日志的索引
						.setLastLogTerm(lastLogId.getTerm()) // 最后一条日志的任期
						.build();
				// 发送预投票请求，当收到响应后则会回调 done
				// this.rpcService.preVote(peer.getEndpoint(), done.request, done);
			}

			// 给自己投一票，虽然节点当前还没有把自己更新为 Candidate 状态，在预投票通过后，正式进行 Leader 选举时才会把自己更新为 Candidate 状态，然后投自己一票。但是预投票流程和正式选举流程差不多，所以也投自己一票
			this.prevVoteCtx.grant(this.nodeId.getPeerId());
			// 判断是否超过集群中半数节点的投票
			if (this.prevVoteCtx.isGranted()) {
				// 正式选举的入口一定持有着写锁，正式选举完成后会释放写锁，preVote 的 finally 就不需要释放写锁了
				doUnlock = false;
				// 如果票数超过了集群半数节点，意味着预投票成功了，开始正式进行投票。下面的这个方法就是正式进行 Leader 选举的方法
				electSelf();
			}
		} finally {
			if (doUnlock) {
				this.writeLock.unlock();
			}
		}
	}

	/** 其它节点处理候选者发送过来的预投票请求，并给出响应 */
	@Override
	public RpcRequests.RequestVoteResponse handlePreVoteRequest(final RpcRequests.RequestVoteRequest request) {
		// 检查正在处理预投票请求的节点是否处于活跃状态
		if (!this.state.isActive()) {
			logger.warn("Node {} is not in active state, currTerm={}.", getNodeId(), this.term);
			return null;
//			return RpcFactoryHelper.responseFactory()
//					.newResponse(com.alipay.sofa.jraft.rpc.RpcRequests.RequestVoteResponse.getDefaultInstance(), RaftError.EINVAL,
//							"Node %s is not in active state, state %s.", getNodeId(), this.state.name());
		}

		final PeerId candidateId = new PeerId();
		// 先从请求中获取候选者节点信息，也就是发送预投票请求的节点信息
		if (!candidateId.parse(request.getServerId())) {
			logger.warn("Node {} received PreVoteRequest from {} serverId bad format.", getNodeId(), request.getServerId());
			// 解析失败则发送失败响应
			return null;
//			return RpcFactoryHelper.responseFactory()
//					.newResponse(com.alipay.sofa.jraft.rpc.RpcRequests.RequestVoteResponse.getDefaultInstance(), RaftError.EINVAL,
//							"Parse candidateId failed: %s.", request.getServerId());
		}

		// 定义一个投票变量，false 表示处理预投票请求的节点还没有给候选者（发送预投票请求的节点）投票
		boolean granted = false;
		do { // 在循环里处理
			// 判断候选者节点是否在当前的集群配置中
			if (!this.conf.contains(candidateId)) {
				logger.warn("Node {} ignore PreVoteRequest from {} as it is not in conf <{}>.", getNodeId(), request.getServerId(), this.conf);
				break;
			}

			// 当前节点（处理预投票请求的节点）判断它所记录的 Leader 是否为空，如果不为空，则说明在当前节点来看 Leader 还在。接着会判断当前节点最后一次收到 Leader 的心跳是否超过阈值，如果没有超过阈值，则说明当前节点记录的 Leader 是有效的，此时会忽略候选者节点的预投票请求
			// 出现这种情况，可能是因为网络延迟，有的 Follower 还没有及时接收到 Leader 的心跳就转变成候选者开始进行预投票活动了
			if (this.leader != null && isCurrentLeaderValid()) {
				logger.info("Node {} ignore PreVoteRequest from {}, term={}, currTerm={}, because the leader {}'s lease is still valid.", getNodeId(), request.getServerId(), request.getTerm(), this.term, this.leader);
				break;
			}

			// 候选者的任期小于当前节点的任期，忽略预投票请求
			if (request.getTerm() < this.term) {
				logger.info("Node {} ignore PreVoteRequest from {}, term={}, currTerm={}.", getNodeId(), request.getServerId(), request.getTerm(), this.term);
				// 这个方法在第一版本不用关注，因为我根本就没做实现
				// checkReplicator(candidateId);
				break;
			}

			// 再次进行了一次复制器检查
			// checkReplicator(candidateId);

			// 获取当前节点（处理预投票请求的节点）的最后一条日志的 ID
			// final LogId lastLogId = this.logManager.getLastLogId(true);
			// 从预投票请求中获得候选者节点的最后一条日志的 ID
			final LogId requestLastLogId = new LogId(request.getLastLogIndex(), request.getLastLogTerm());

			// 如果候选者节点的最后一条日志的 ID 和任期大于当前节点，当前节点就给候选者投票
			// granted = requestLastLogId.compareTo(lastLogId) >= 0;
			granted = true;
			logger.info("Node {} received PreVoteRequest from {}, term={}, currTerm={}, granted={}, requestLastLogId={}.", getNodeId(), request.getServerId(), request.getTerm(), this.term, granted, requestLastLogId);
		} while (false);

		// 构建给候选者节点返回的响应
		return RpcRequests.RequestVoteResponse.newBuilder()
				.setTerm(this.term)
				.setGranted(granted) // true 表示当前节点给候选者投票了
				.build();
	}

	/** 候选者节点向其它节点发送了索要预投票的请求，收到其它节点的响应后就会调用这个方法。term 参数代表的是当前候选者节点发送索要投票请求之前的任期，这个逻辑可以从 OnPreVoteRpcDone#run 方法中查看 */
	public void handlePreVoteResponse(final PeerId peerId, final long term, final RpcRequests.RequestVoteResponse response) {
		// 当前节点（以 FOLLOWING 身份）向其它节点发送预投票请求后，并收到其它节点的响应，当前节点就应该还是 FOLLOWING，才会继续执行后面的代码。如果不是 FOLLOWING，就直接退出
		// 1. 当前节点不可能是 Leader
		// 2. 当前节点不可能是 Candidate，因为只有只有正式选举时才会变为 Candidate，现在还是预投票
		if (this.state != State.FOLLOWING) {
			logger.warn("Node {} received invalid PreVoteResponse from {}, state not in STATE_FOLLOWER but {}.", getNodeId(), peerId, this.state);
			return;
		}

		// 当前节点收到预投票的响应时的前后任期如果不一致则直接退出该方法
		if (term != this.term) {
			logger.warn("Node {} received invalid PreVoteResponse from {}, term={}, currTerm={}.", getNodeId(), peerId, term, this.term);
			return;
		}

		// 响应预投票的节点的任期 < 发起预投票请求的节点的任期，说明响应者赞成发起投票者成为 Leader
		// 响应预投票的节点的任期 = 发起预投票请求的节点的任期，还要比对二者的 log index
		// 响应预投票的节点的任期 > 发起预投票请求的节点的任期，说明响应者反对发起投票者成为 Leader，发起预投票的节点身份降级为 FOLLOWER
		if (response.getTerm() > this.term) {
			logger.warn("Node {} received invalid PreVoteResponse from {}, term {}, expect={}.", getNodeId(), peerId, response.getTerm(), this.term);
			// TODO 执行身份降级，为什么呢？
			stepDown(response.getTerm(), false, null);
			return;
		}
		logger.info("Node {} received PreVoteResponse from {}, term={}, granted={}.", getNodeId(), peerId, response.getTerm(), response.getGranted());

		// 从响应中判断回复响应的节点是否给当前节点投票了
		if (response.getGranted()) {
			//如果投票了，就在这里收集票数到计票器中
			this.prevVoteCtx.grant(peerId);
			// 如果获得票数超过集群节点一半，就进入正式选举阶段
			if (this.prevVoteCtx.isGranted()) {
				// 开始正式投票
				electSelf();
			}
		}
	}

	/** 这个类的对象中封装着一个回调方法，该方法会在候选者节点接收到预投票的响应之后被回调 */
	private class OnPreVoteRpcDone extends RpcResponseClosureAdapter<RpcRequests.RequestVoteResponse> {

		final long         startMs;
		final PeerId       peer;
		final long         term;
		RpcRequests.RequestVoteRequest request;

		public OnPreVoteRpcDone(final PeerId peer, final long term) {
			super();
			this.startMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
			this.peer = peer;
			this.term = term;
		}

		@Override
		public void run(final Status status) {
//			NodeImpl.this.metrics.recordLatency("pre-vote", Utils.monotonicMs() - this.startMs);
//			if (!status.isOk()) {
//				LOG.warn("Node {} PreVote to {} error: {}.", getNodeId(), this.peer, status);
//			} else {
//				handlePreVoteResponse(this.peer, this.term, getResponse());
//			}
		}
	}
	// endregion

	// region 正式选举投票
	/** 在规定的时间内还未正式选举出 Leader，则会执行下面的代码 */
	private void handleVoteTimeout() {
		// 只有 CANDIDATE 节点才会发起正式 Leader 选举，即，FOLLOWING 触发超时选举之后，先进行预投票操作，预投票通过后，才会进入正式的 Leader 选举（electSelf 方法，会将节点的状态改为 CANDIDATE）
		// 当选举一直没有成功（发生了网络分区或有多个 CANDIDATE），就会一直让 CANDIDATE 节点的选举超时定时器工作
		if (this.state != State.CANDIDATE) {
			return;
		}

		// 预投票的目的是判断是否发生了网络分区，从而导致发起预投票的节点任期无限增长。当进入正式投票时，也会有可能预投票时没有网络分区，但正式投票时发生网络分区，也会导致任期无限增长。因此，softjraft 提供了一个选项；
		// ● isStepDownWhenVoteTimedout = true（默认为 true）时，选举超时发生时（与其它节点失联或没有得到足够的选票），会调用 `stepDown` 方法使得节点从 `CANDIDATE` 降级为 `FOLLOWING（再次开启超时选举定时器）`，然后再执行预投票，预投票成功后再次进入正式投票；
		// ● isStepDownWhenVoteTimedout = false 时，选举超时发生时（与其它节点失联或没有得到足够的选票），直接进入下一轮正式选举，任期的确会一直增加；
		// isStepDownWhenVoteTimedout 通常是配置在全局的 Options 中
		boolean isStepDownWhenVoteTimedout = true;
		if (isStepDownWhenVoteTimedout) {
			logger.warn("Candidate node {} term {} steps down when election reaching vote timeout: fail to get quorum vote-granted.", this.nodeId, this.term);
			// 进行身份降级，把当前节点降级为 Follower，然后重新开启超时选举定时器
			stepDown(this.term, false, null);
			// 再次进行预投票操作
			preVote();
		} else {
			logger.debug("Node {} term {} retry to vote self.", getNodeId(), this.term);
			// 如果没有设置身份降级，就直接再次进行选举操作，这里开启了新一轮选举操作
			electSelf();
		}
	}

	/** 当前节点发起正式 Leader 选举（向其它节点发起正式的索票请求），集群中的每一个节点在一个任期内只允许投一票，该方法的大部分逻辑和预投票的逻辑差不多 */
	private void electSelf() {
		try {
			logger.info("Node {} start vote and grant vote self, term={}.", getNodeId(), this.term);
			// 当前节点若不在集群中则直接退出
			if (!this.conf.contains(this.nodeId.getPeerId())) {
				logger.warn("Node {} can't do electSelf as it is not in {}.", getNodeId(), this.conf);
				return;
			}

			// 开始修改 state 为 CANDIDATE，自增 term，这些涉及到修改共享变量，所以肯定是在持有锁的状态下做的这些操作

			//~ 1. FOLLOWING 启动超时选举，预投票成功后，开始正式投票。FOLLOWING 状态时启动的超时选举定时器就没有意义了，需要将其停止
			if (this.state == State.FOLLOWING) {
				logger.debug("Node {} stop election timer, term={}.", getNodeId(), this.term);
				// 停止超时选举定时器
				this.electionRepeated.stop();
			}

			// 既然正在选举，肯定是没有 Leader 了，所以在进行选举时把当前节点保存的 Leader 节点信息设置为空
			updateLeaderId(null, new Status(RaftError.ERAFTTIMEDOUT, "A follower's leader_id is reset to NULL as it begins to request_vote."));

			//~ 2. 将当前节点状态更改为 CANDIDATE
			this.state = State.CANDIDATE;

			// 递增当前节点的任期
			this.term++;
			// 当前节点在当前任期只能投一票，要为自己投票，所以把为谁投票的 ID 设置为自己的服务器 ID
			this.votedId = this.nodeId.getPeerId();
			// 记录一条日志，表示要启动投票超时计时器了
			logger.debug("Node {} start vote timer, term={} .", getNodeId(), this.term);

			//~ 3. 启动选举超时定时器（预投票成功后开始正式 Leader 选举，在规定的时间内仍未选举出 Leader，则会进入下一轮 Leader 选举）
			this.voteRepeated.start();

			// 初始化正式投票计数器
			this.voteCtx.init(this.conf, this.conf);


			// 使用日志管理器来获得最后一条日志的 ID：final LogId lastLogId = this.logManager.getLastLogId(true);
			final LogId lastLogId = new LogId(0,0);

			// 遍历集群中的节点，向各个节点发送正式的索票请求
			for (final PeerId peer : this.conf.getPeers()) {
				if (peer.equals(this.nodeId.getPeerId())) { // 跳过当前节点
					continue;
				}

				// 连接到遍历到的这个节点，如果不能成功连接，就记录日志
//			if (!this.rpcService.connect(peer.getEndpoint())) {
//				logger.warn("Node {} channel init failed, address={}.", getNodeId(), peer.getEndpoint());
//				continue;
//			}

				// 创建一个回调对象，这个对象中实现了一个回调函数，该回调函数会在接收到正式投票请求的节点的的响应后被回调
				final OnRequestVoteRpcDone done = new OnRequestVoteRpcDone(peer, this.term, this);
				// 创建正式投票请求
				done.request = RpcRequests.RequestVoteRequest.newBuilder()
						.setPreVote(false) // 设置成 false 意味着不是预投票请求
						.setGroupId(this.nodeId.getGroupId())
						.setServerId(this.nodeId.getPeerId().toString()) // 发起正式投票（索票）的节点
						.setPeerId(peer.toString()) // 当前节点发起的正式投票（索票）请求发送给哪个节点
						.setTerm(this.term)
						.setLastLogIndex(lastLogId.getIndex()) // 发起正式投票（索票）请求的节点的最后一条日志的索引
						.setLastLogTerm(lastLogId.getTerm()) // 发起正式投票（索票）请求的节点的最后一条日志所在的任期
						.build();
				// 发送请求
				// this.rpcService.requestVote(peer.getEndpoint(), done.request, done);
			}

			// 当前节点正在申请成为领导者，任期和投票记录（投给自己）有变更了，所以需要元数据持久化
			this.metaStorage.setTermAndVotedFor(this.term, this.nodeId.getPeerId());
			// 在这里给当前节点投票
			this.voteCtx.grant(this.nodeId.getPeerId());

			// 如果获取的投票已经超过半数，则让当前节点成为领导者
			if (this.voteCtx.isGranted()) {
				// becomeLeader();
			}
		} finally {
			this.writeLock.unlock();
		}
	}

	/** 当前节点接收到 CANDIDATE 节点发送过来的正式投票（索票）请求时会调用这个方法进行处理，正式投票（索票）的请求就是该方法中的参数对象 */
	@Override
	public RpcRequests.RequestVoteResponse handleRequestVoteRequest(final RpcRequests.RequestVoteRequest request) {
		this.writeLock.lock();
		try {
			// 检查当前节点（处理正式投票（索票）请求的节点）是否处于活跃状态
			if (!this.state.isActive()) {
				logger.warn("Node {} is not in active state, currTerm={}.", getNodeId(), this.term);
				return null;
				// return RpcFactoryHelper.responseFactory().newResponse(RpcRequests.RequestVoteResponse.getDefaultInstance(), RaftError.EINVAL, "Node %s is not in active state, state %s.", getNodeId(), this.state.name());
			}

			// 创建一个 PeerId 对象，用来解析 CANDIDATE 节点信息的
			final PeerId candidateId = new PeerId();
			// 解析 CANDIDATE 节点信息
			if (!candidateId.parse(request.getServerId())) {
				logger.warn("Node {} received RequestVoteRequest from {} serverId bad format.", getNodeId(), request.getServerId());
				return null;
				// return RpcFactoryHelper.responseFactory().newResponse(RpcRequests.RequestVoteResponse.getDefaultInstance(), RaftError.EINVAL, "Parse candidateId failed: %s.", request.getServerId());
			}

			do {
				// CANDIDATE 节点的日志是否比当前节点最后一条日志新
				boolean logCandidateNewerThanCurrent = true;

				if (request.getTerm() > this.term) {
					// CANDIDATE 节点的任期比当前节点的任期大，当前节点可能是 FOLLOWING，也可能是 CANDIDATE（同时出现了多个候选者），那么让当前节点（任期较小）身份降级（相比较于发送正式索票请求的那个 CANDIDATE 节点，当前节点任期更小，还是让任期大的节点成为 Leader 吧）
					// stepDown 方法会让当前节点的任期变更为 CANDIDATE 节点的更为大的任期，也会把当前节点的投票记录重置
					stepDown(request.getTerm(), false, null);
				} else if (request.getTerm() == this.term) {
					// CANDIDATE 节点的任期和当前节点的任期相等，谁也不必改变谁的，只等着 Leader 选出来，让 Leader 给 CANDIDATE 发送信息，让 CANDIDATE 身份降级即可
					// 任期相等，继续比较日志

					// 获取当前节点最后一条日志：final LogId lastLogId = this.logManager.getLastLogId(true);
					LogId lastLogId = new LogId(0,0);
					// 校验 CANDIDATE 节点的日志是否比当前节点最后一条日志新
					logCandidateNewerThanCurrent = new LogId(request.getLastLogIndex(), request.getLastLogTerm()).compareTo(lastLogId) >= 0;
					if (logCandidateNewerThanCurrent) {
						// CANDIDATE 节点和当前节点的任期相等，但 CANDIDATE 节点的日志比当前节点新，那么让当前节点（日志较旧）身份降级（相比较于发送正式索票请求的那个 CANDIDATE 节点，当前节点日志更旧，还是让日志新的节点成为 Leader 吧）
						// stepDown 方法会让当前节点的任期变更为 CANDIDATE 节点的更为大的任期，也会把当前节点的投票记录重置
						stepDown(request.getTerm(), false, null);
					}
				} else {
					// CANDIDATE 节点的任期比当前节点的任期小，直接忽略此请求
					logger.info("Node {} ignore RequestVoteRequest from {}, term={}, currTerm={}.", getNodeId(), request.getServerId(), request.getTerm(), this.term);
					break;
				}

				// stepDown 方法会把当前节点的投票记录重置，且同一个任期只能投一票。当投票记录为空时才会给 CANDIDATE 投票
				if (this.votedId == null) {
					// 记录当前节点的投票信息
					this.votedId = candidateId;
					// 投票信息更改后需要将元数据信息持久化
					this.metaStorage.setVotedFor(candidateId);
				}
			} while (false);

			// 这里构建响应对象，会调用 setGranted(boolean) 方法，boolean 参数表示是否投票给对方的意思
			return RpcRequests.RequestVoteResponse.newBuilder()
					.setTerm(this.term)
					// TODO request.getTerm() == this.currTerm，这里大家可能会疑惑，为什么要再次判断一下请求中的任期是否和节点的当前任期是否相等？
					.setGranted(request.getTerm() == this.term && candidateId.equals(this.votedId))
					.build();
		} finally {
			this.writeLock.unlock();
		}
	}

	/** 如果当前节点的身份是 CANDIDATE，发送了索票请求给其它节点后，收到其它节点的响应后就会调用这个方法。peerId 为 CANDIDATE 节点将索票请求发给谁，term 表示 CANDIDATE 节点发送索票请求时所携带的任期，这个逻辑可以从 {@link OnRequestVoteRpcDone#run} 方法中查看 */
	public void handleRequestVoteResponse(final PeerId peerId, final long term, final RpcRequests.RequestVoteResponse response) {
		// 当前候选者节点接收到响应之后，发现自己的身份不是 CANDIDATE 了，直接退出该方法。也许在接收到投票响应之前已经收到了集群中选举出来的 Leader 的信息，这样就会把自己的身份降级为 Follower 了
		if (this.state != State.CANDIDATE) {
			logger.warn("Node {} received invalid RequestVoteResponse from {}, state not in STATE_CANDIDATE but {}.", getNodeId(), peerId, this.state);
			return;
		}

		// 判断当前候选者节点的任期是否发生了变化，方法参数中的 term 代表的是当前候选者节点发送索票请求时的任期，而在发送索票请求之后，收到响应之前，很可能任期就被新的 Leader 改变了，如果前后任期不一致，直接退出该方法
		if (term != this.term) {
			logger.warn("Node {} received stale RequestVoteResponse from {}, term={}, currTerm={}.", getNodeId(), peerId, term, this.term);
			return;
		}

		// 如果接收到响应后发现对方的任期比自己大，说明对方投的拒绝票，此时直接退出该方法，并让自己身份降级为 Follower
		if (response.getTerm() > this.term) {
			logger.warn("Node {} received invalid RequestVoteResponse from {}, term={}, expect={}.", getNodeId(), peerId, response.getTerm(), this.term);
			// 身份降级
			// stepDown((int) response.getTerm(), false, new Status(RaftError.EHIGHERTERMRESPONSE, "Raft node receives higher term request_vote_response."));
			stepDown(response.getTerm(), false, null);
			return;
		}

		// 从请求中判断对方是否投了赞成票
		if (response.getGranted()) {
			// 如果收到了投票就把来自 peerId 节点的投票收集到计票器中
			this.voteCtx.grant(peerId);
			// 使用投票计算器计算当前节点收到的投票是否超过了集群中半数节点
			if (this.voteCtx.isGranted()) {
				// 超过了就让当前节点成为领导者
				// becomeLeader();
			}
		}
	}

	/** 该类的对象中封装着一个回调方法，该方法会在候选者节点（发起正式投票请求的那个节点）接收到其它节点的响应之后被回调 */
	private class OnRequestVoteRpcDone extends RpcResponseClosureAdapter<RpcRequests.RequestVoteResponse> {
		final long         startMs;
		/** CANDIDATE 节点发送正式投票（索取）请求给哪个节点 */
		final PeerId       peer;
		/** CANDIDATE 节点发送索票请求时所携带的 CANDIDATE 的任期 */
		final long         term;
		/** CANDIDATE 节点对象 */
		final NodeImpl     node;
		RpcRequests.RequestVoteRequest request;

		public OnRequestVoteRpcDone(final PeerId peer, final long term, final NodeImpl node) {
			super();
			this.startMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
			this.peer = peer;
			this.term = term;
			this.node = node;
		}

		@Override
		public void run(final Status status) {
//			NodeImpl.this.metrics.recordLatency("request-vote", Utils.monotonicMs() - this.startMs);
//			if (!status.isOk()) {
//				LOG.warn("Node {} RequestVote to {} error: {}.", this.node.getNodeId(), this.peer, status);
//			} else {
//				this.node.handleRequestVoteResponse(this.peer, this.term, getResponse());
//			}
		}
	}
	// endregion

	// region 当选为 Leader
	/** 当前节点成为 Leader */
	private void becomeLeader() {
		// 判断当前节点状态是否为 CANDIDATE，LEADING 节点状态只能从 CANDIDATE 变更过来
		Asserts.isTrue(this.state == State.CANDIDATE, "Illegal state: " + this.state);
		// 当前节点正在成为 Leader，记录这条日志
		logger.info("Node {} become leader of group, term={}, conf={}, oldConf={}.", getNodeId(), this.term, this.conf, this.conf);
		// 将当前节点状态设置为 LEADING
		this.state = State.LEADING;
		// 设置 Leader 为当前服务器的 ID，即 Leader 就是自己
		this.leader = this.nodeId.getPeerId();
		// 重置 raft 复制组的 term。Leader 一旦确定了，其它节点的信息都会封装到 ReplicatorGroup 复制器组对象中，Leader 的日志都会通过复制组发送给每一个 Follower 节点，Follower 会被封装复制组器 ReplicatorGroup 中的一个个的复制器 Replicator 对象中
		this.replicatorGroup.resetTerm(this.term);

		// 遍历集群配置信息中的各个节点，将各个 Follower 封装为复制器 Replicator 对象
		for (final PeerId peer : this.conf.getPeers()) {
			// 如果遍历到的节点就是当前节点，直接跳过这次循环
			if (peer.equals(this.nodeId.getPeerId())) {
				continue;
			}
			// 接下来要向复制器组中添加复制器对象了，记录一条日志
			logger.debug("Node {} add a replicator, term={}, peer={}.", getNodeId(), this.term, peer);
			// addReplicator 方法中，Follower 节点会被包装成一个个复制器 Replicator 对象，然后把每一个复制器对象添加到这个复制组器对象中，并且在 addReplicator 方法内部，还会启动心跳定时器，以及发送探针请求。此时 Leader 会向集群中的所有 Follower 发送信息了
			if (!this.replicatorGroup.addReplicator(peer)) {
				// 如果添加复制器失败，记录错误日志
				logger.error("Fail to add a replicator, peer={}.", peer);
			}
		}

		// 和学习者有关的操作，暂时不用关注
		for (final PeerId peer : this.conf.getPeers()) {
			logger.debug("Node {} add a learner replicator, term={}, peer={}.", getNodeId(), this.term, peer);
			// 这里的 ReplicatorType.Learner 枚举类对象表示了当前包装为复制器的节点身份为学习者
			if (!this.replicatorGroup.addReplicator(peer, ReplicatorType.Learner)) {
				logger.error("Fail to add a learner replicator, peer={}.", peer);
			}
		}

//		if (this.confCtx.isBusy()) {
//			throw new IllegalStateException();
//		}
//		// 刷新配置
//		this.confCtx.flush(this.conf.getConf(), this.conf.getOldConf());
	}
	// endregion

	// region 当前节点接收到 Leader 发送过来心跳、探针、复制日志的消息的处理逻辑
	/** 当前节点接收到 Leader 发送过来的消息时，不管消息是日志复制还是心跳消息，还是探针消息，都会调用该方法进行处理。其实 Leader 发送的日志复制消息，以及心跳消息探针消息，使用的都是同一个 RPC 请求 */
	@Override
	public Message handleAppendEntriesRequest(final RpcRequests.AppendEntriesRequest request, final RpcRequestClosure done) {
		// 是否释放写锁
		boolean doUnlock = true;
		this.writeLock.lock();

		// 从请求中获得 Leader 传送过来的日志的条数
		final int entriesCount = request.getEntriesCount();
		try {
			// 判断当前节点是否活跃
			if (!this.state.isActive()) {
				logger.warn("Node {} is not in active state, currTerm={}.", getNodeId(), this.term);
				// return RpcFactoryHelper.responseFactory().newResponse(RpcRequests.AppendEntriesResponse.getDefaultInstance(), RaftError.EINVAL, "Node %s is not in active state, state %s.", getNodeId(), this.state.name());
			}

			// 解析出 Leader 的端点信息
			final PeerId serverId = new PeerId();
			if (!serverId.parse(request.getServerId())) {
				logger.warn("Node {} received AppendEntriesRequest from {} serverId bad format.", getNodeId(), request.getServerId());
				// return RpcFactoryHelper.responseFactory().newResponse(RpcRequests.AppendEntriesResponse.getDefaultInstance(), RaftError.EINVAL, "Parse serverId failed: %s.", request.getServerId());
			}

			// Leader.term < 当前节点的任期，则直接忽略这条请求。如果出现网络分区了，分区1（node1-leader, node2-follower）的任期为 2，分区2（node3-leader，node4-follower）的任期为 5，node1 给 node4 发送心跳，就会出现这种情况，此时 node4 会忽略这个请求
			if (request.getTerm() < this.term) {
				logger.warn("Node {} ignore stale AppendEntriesRequest from {}, term={}, currTerm={}.", getNodeId(), request.getServerId(), request.getTerm(), this.term);
				return RpcRequests.AppendEntriesResponse.newBuilder().setSuccess(false).setTerm(this.term).build();
			}

			// 根据 Leader 的任期来判断当前节点是否需要进行身份降级，即 markdown 中描述的当前节点处理心跳请求的三种情况就是由这个方法处理的，该方法中很可能会把当前节点的 leader 设置为请求中传递过来的节点的 id
			checkStepDown(request.getTerm(), serverId);

			// 当前节点自己记录的 Leader 和发送消息过来的 Leader 并不相同，很可能就是发生了网络分区，分区故障恢复后，另一个分区的 Leader 发送消息过来了
			// TODO 运行到这里，说明 leader.term >= 当前节点的任期，且当前节点记录的 leader 和发送心跳请求的 leader 不一致，此时该怎么处理呢？
			if (!serverId.equals(this.leader)) {
				logger.error("Another peer {} declares that it is the leader at term {} which was occupied by leader {}.", serverId, this.term, this.leader);
				/*
				// 这里给请求中的任期加 1，是为了提高当前节点的任期
				stepDown(request.getTerm() + 1);
				// 在返回的响应中也给任期加1，这样发送心跳过来的领导者就会在接收到响应后对自己的身份降级
				// 如果是另一个领导者发送消息过来，当前节点的任期已经增加1了，那么另一个领导者在接收到响应后也会进行身份降级
				// 集群中没有领导者之后，很快就会触发超时选举了，选举出新的领导者，这就是解决脑裂问题的大概逻辑
				return RpcRequests.AppendEntriesResponse.newBuilder().setSuccess(false).setTerm(request.getTerm() + 1).build();
				 */
			}

			// 更新最后一次接收到 Leader 消息的时间
			this.lastLeaderTimestamp = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());
			logger.info("follower received heartbeat");

			// Leader 发送过来的消息不一定就是心跳，也有可能是复制日志的消息，消息对象中封装着传入过来的日志，因此下面的代码都是和日志有关的
			// 如果当前节点要复制 Leader 的日志，肯定要对比一下当前节点和请求中日志是否符合要求，当前版本的代码会省略这个逻辑，后面会补充
			// TODO 处理复制日志

			// 如果请求中的日志条目为 0，说明当前的请求是心跳或者是探针消息
			if (entriesCount == 0) {
				// TODO 在处理探针消息的逻辑中删除了一些和日志相关的代码，后面会加上

				// 创建响应对象
				final RpcRequests.AppendEntriesResponse.Builder respBuilder = RpcRequests.AppendEntriesResponse.newBuilder()
						.setSuccess(true)
						.setTerm(this.term)
						// 这里暂时写死为 0，本来应该从日志组件中获取当前节点最后一条日志的索引，这样 Leader 收到哦反馈后就知道从哪里开始复制日志了
						.setLastLogIndex(0);

				doUnlock = false;
				this.writeLock.unlock();
				return respBuilder.build();
			}
			return null;
		} finally {
			if (doUnlock) {
				this.writeLock.unlock();
			}
		}
	}

	/** markdown 中所描述的 Follower 收到来自 Leader 的消息时的三种情况的处理方法
	 * @param requestTerm Leader 发送过来的消息中所携带的 Leader 的任期
	 * @param serverId Leader 的端点信息
	 */
	private void checkStepDown(final long requestTerm, final PeerId serverId) {
		// Status 对象用来封装当前节点身份变更的重要信息
		final Status status = new Status();
		if (requestTerm > this.term) {
			// leader.term > 当前节点的任期，当前节点就要变成 Follower
			status.setError(RaftError.ENEWLEADER, "Raft node receives message from new leader with higher term.");
			// 下面的方法中会将当前节点的状态设置为 Follower
			stepDown(requestTerm, false, status);
		} else { // 如果走到这里，意味着 Leader.term <= 当前节点的任期
			if (this.state != State.FOLLOWING) {
				// TODO 当前节点也有可能是 Leader，且两个 Leader 的任期还是一样的，说明出现了脑裂
				// 当前节点是候选者节点，并且收到了和自己任期相同的领导者的信息
				status.setError(RaftError.ENEWLEADER, "Candidate receives message from new leader with the same term.");
				// 让自己降级为 Follower，集群中有多个候选者时可能会出现这种情况
				stepDown(requestTerm, false, status);
			} else if (this.leader == null) {
				// 当前节点是 Follower，并且自己记录的 leader 信息为空，现在收到到了 Leader 发送过来的信息，说明 Leader 是刚刚选举出来，此时让自己保持 FOLLOWING 状态即可
				status.setError(RaftError.ENEWLEADER, "Follower receives message from new leader with the same term.");
				stepDown(requestTerm, false, status);
			}
		}

		// 设置当前节点记录的 leader 信息
		if (this.leader == null) {
			updateLeaderId(serverId, status);
		}
	}
	// endregion
}
